quote_sqlid <- function(con, what) {

  if (length(what) == 1L && what == "*") {
    what
  } else {
    DBI::dbQuoteIdentifier(con, what)
  }
}

#' Select from a table
#'
#' NOTICE: there is no safety check of argument where in select_from_where()
#'
#' @param con a DBIConnection object
#' @param tbl name of table to query
#' @param what what variables/columns to query
#' @param where a string of the where clause (without "WHERE")
#' @param ... named arguments, parsed as WHERE AND clause
#'
#' @return data.table
#' @export
#'
db_select_from <- function(con, tbl, what = "*", ...) {

  tbl  <- quote_sqlid(con, tbl)
  what <- quote_sqlid(con, what)
  q <- sprintf('SELECT %s FROM %s',
               paste0(what, collapse = ','),
               tbl)

  nwhere <- 0L
  if (...length()) {
    where <- data.table::CJ(..., unique = TRUE)
    nwhere <- nrow(where)
  }
  if (nwhere) {
    #construct where clause
    var <- seq_along(where)
    var_quoted <- quote_sqlid(con, names(where))
    where_clause <- sprintf("%s = $%d", var_quoted, var)
    q <- paste(q, "WHERE", paste0(where_clause, collapse = " AND "))
  }

  ans <- tryCatch({
    if (nwhere) {
      r <- DBI::dbSendQuery(con, q)
      on.exit(DBI::dbClearResult(r))
      DBI::dbBind(r, unname(as.list(where)))
      DBI::dbFetch(r)
    } else {
      DBI::dbGetQuery(con, q)
    }
  },
  error = function(e)
  {
    warning(toString(e), call. = FALSE)
    NULL
  })

  data.table::setDT(ans)
}

#' @rdname db_select_from
#' @export
#'
db_select_where <- function(con, tbl, what = "*", where = "") {

  what <- quote_sqlid(con, what)
  tbl  <- quote_sqlid(con, tbl)
  q <- sprintf('SELECT %s FROM %s', paste0(what, collapse = ','), tbl)

  if (nzchar(where)) {
    q <- paste(q, "WHERE", where)
  }

  ans <- tryCatch({
    DBI::dbGetQuery(con, q)
  },
  error = function(e)
  {
    warning(toString(e), call. = FALSE)
    NULL
  })

  data.table::setDT(ans)
}

#' Insert data to a table
#'
#' @param con a DBIConnection object
#' @param tbl name of table to insert to
#' @param dt data to insert
#' @param conflict how to deal with conflict
#' @param constraint constrained columns, only used when conflict == "replace" and connection is to a PostgreSQL database
#'
#' @return number of rows inserted
#' @export
#'
db_insert_to <- function(con, tbl, dt, conflict = c("default", "replace", "ignore"), constraint = data.table::key(dt)) {

  conflict <- match.arg(conflict)
  tbl <- quote_sqlid(con, tbl)
  tbl_temp <- quote_sqlid(con, paste0(tbl, "_tmp_", digest::digest(dt, algo = "xxhash32")))

  cols <- quote_sqlid(con, names(dt))
  cols_sql <- paste0(cols, collapse = ", ")

  verb <- ""
  target_template <- "INTO %s(%s) SELECT * FROM %s"
  suffix <- ""

  if (methods::is(con, "SQLiteConnection")) {
    # SQLite
    verb <- switch(conflict,
                   default = "INSERT",
                   replace = "INSERT OR REPLACE",
                   ignore  = "INSERT OR IGNORE")
  } else if (methods::is(con, "PqConnection")) {
    # PostgreSQL
    verb <- "INSERT"
    if (conflict == "replace")
    {
      if (is.null(constraint))
      {
        warning("Unique/exclusion constraint is required to perform update.", call. = FALSE)
        return(0L)
      }
      cstr <- paste0(quote_sqlid(con, constraint), collapse = ", ")
      rplc <- paste0(sprintf("%s = EXCLUDED.%s", cols, cols), collapse = ", ")
      suffix <- sprintf("ON CONFLICT (%s) DO UPDATE SET %s", cstr, rplc)
    }
    else if (conflict == "ignore")
    {
      suffix <- "ON CONFLICT DO NOTHING"
    }
  } else if (methods::is(con, "MariaDBConnection")) {
    # MariaDB
    verb <- switch(conflict,
                   default = "INSERT",
                   replace = "REPLACE",
                   ignore  = "INSERT IGNORE")
  } else {
    # Other
    cls <- class(con)
    if (conflict == "default")
    {
      verb <- "INSERT"
    }
    else
    {
      warning("Don't know how to ", conflict, " conflicts for ", cls, call. = FALSE)
      return(0L)
    }
  }
  # %verb% INTO %s(%s) SELECT * FROM %s %suffix%
  q_template <- paste(verb, target_template, suffix)
  q <- sprintf(q_template, tbl, cols_sql, tbl_temp)

  r <- tryCatch({
    DBI::dbBegin(con)
    # Write to temp table
    DBI::dbWriteTable(conn = con, name = tbl_temp, value = dt,
                      overwrite = TRUE, temporary = TRUE, copy = TRUE)
    # Insert from temp table
    r <- DBI::dbExecute(con, q)
    DBI::dbCommit(con)
    r
  }, error = function(e) {
    DBI::dbRollback(con)
    warning(toString(e), call. = FALSE)
    0L
  })

  r
}

#' Create an empty table
#'
#' @param con a DBIConnection object
#' @param tbl table name
#' @param dt reference data.frame
#'
#' @return TRUE/FALSE
#' @export
#'
db_create_table <- function(con, tbl, dt) {

  dt <- dt[0L, , drop = FALSE]
  r <- tryCatch({
    DBI::dbWriteTable(con, tbl, dt)
    TRUE
  }, error = function(e) {
    warning(toString(e), call. = FALSE)
    FALSE
  })

  r
}

#' Create index
#'
#' @param con a DBIConnection object
#' @param idx index name
#' @param tbl table name
#' @param col columns to create index on
#' @param ASC index order
#' @param unique whether to create a unique index
#'
#' @return TRUE/FALSE
#' @export
#'
db_create_index <- function(con, idx, tbl, col, ASC = NULL, unique = FALSE) {

  tbl <- quote_sqlid(con, tbl)
  col <- quote_sqlid(con, col)
  idx <- quote_sqlid(con, idx)

  if (unique) {
    unique = "UNIQUE"
  } else {
    unique <- ""
  }

  n_order <- length(ASC)
  if (n_order) {
    if (n_order == 1L || n_order == length(col))
    {
      order <- ifelse(ASC, "ASC", "DESC")
    }
    else
    {
      warning("ASC should be of length 1 or same as col.", call. = FALSE)
      return(FALSE)
    }
  } else {
    order <- ""
  }
  col <- paste(col, order)

  q <- sprintf("CREATE %s INDEX %s ON %s (%s)",
               unique,
               idx,
               tbl,
               paste0(col, collapse = ", "))

  r <- tryCatch({
    DBI::dbBegin(con)
    DBI::dbExecute(con, q)
    DBI::dbCommit(con)
    TRUE
  }, error = function(e) {
    DBI::dbRollback(con)
    warning(toString(e), call. = FALSE)
    FALSE
  })

  r
}
